/**
 * 
 */
package com.googlecode.xcarve.netty.core;

import io.netty.util.internal.chmv8.ConcurrentHashMapV8;

import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.googlecode.xcarve.common.config.CarveConstances;
import com.googlecode.xcarve.object.RPCReponse;
import com.googlecode.xcarve.object.RPCRequest;

/**
 * @author xinyun.d
 * 2014年9月16日
 */
public class DoveFuture implements RpcFuture{
	static final Logger logger = LoggerFactory.getLogger(DoveFuture.class);
	private final long seqId;
	private final long timeout;
	private final Lock lock = new ReentrantLock();
	private final Condition done = lock.newCondition();
	private volatile RPCReponse response;
	private long sendTime;
	private static final Map<Long,DoveFuture> future = new ConcurrentHashMapV8<Long,DoveFuture>();
	
	public DoveFuture(RPCRequest request){
		seqId = request.getSeqId();
		timeout = request.getTimeout()<=0 ? CarveConstances.DEFAULT_TIMEOUT : request.getTimeout();
		future.put(seqId, this);
	}
	
	/* (non-Javadoc)
	 * @see com.dove.transport.netty.core.RpcFuture#isDone()
	 */
	@Override
	public boolean isDone() {
		return this.response!=null;
	}

	/* (non-Javadoc)
	 * @see com.dove.transport.netty.core.RpcFuture#get()
	 */
	@Override
	public RPCReponse get() throws Exception {
		return get(CarveConstances.DEFAULT_TIMEOUT);
	}

	/* (non-Javadoc)
	 * @see com.dove.transport.netty.core.RpcFuture#get(long)
	 */
	@Override
	public RPCReponse get(long timeout) throws Exception {
		if(timeout<=0){
			timeout = CarveConstances.DEFAULT_TIMEOUT;
		}
		if(!isDone()){
			long start = System.currentTimeMillis();
			lock.lock();
			try {
				while(!isDone()){
					done.await(timeout, TimeUnit.MILLISECONDS);
					if(isDone() || System.currentTimeMillis() - start > timeout){
						break;
					}
				}
			} catch (Exception e) {
				throw e;
			}finally{
				lock.unlock();
			}
			if(!isDone()){
				throw new TimeoutException("Waiting server-side response message timeout");
			}
		}
		return response;
	}

	/* (non-Javadoc)
	 * @see com.dove.transport.netty.core.RpcFuture#invokeCallback()
	 */
	@Override
	public void invokeCallback() {
		
	}
	
	public static void readResponse(final RPCReponse response){
		DoveFuture dfuture = future.remove(response.getSeqId());
		if(dfuture!=null){
			dfuture.doHandReceive(response);
		}
	}
	
	public void doHandReceive(final RPCReponse response){
		lock.lock();
		try {
			this.response = response;
			if(done != null){
				done.signal();
			}
		} catch (Exception e) {
		}finally{
			lock.unlock();
		}
	}
	
	public void cancelFutrure(){
		 RPCReponse response = new RPCReponse(seqId);
		 response.setMessage("request future has been canceled.");
		 response.setStatus(false);
         this.response = response ;
         future.remove(seqId);
	}

	public Object returnResult(){
		RPCReponse res = this.response;
		if(res == null){
			throw new IllegalStateException("return response cannot be null.");
		}
		if(res.isStatus() == true){
			return res.getResult();
		}
		throw new RuntimeException(res.getMessage());
	}
	
	public void setResponse(RPCReponse response){
		this.response = response;
	}
	
	/**
	 * @param sendTime the sendTime to set
	 */
	public void setSendTime() {
		this.sendTime = System.currentTimeMillis();
	}
	
	public static void initSendTime(long seqId){
		DoveFuture dove = future.get(seqId);
		if(dove!=null){
			dove.setSendTime();
		}
	}
	
	public long getStartTime(){
		return this.sendTime;
	}
	
	  /**
	 * @return the seqId
	 */
	public long getSeqId() {
		return seqId;
	}

	/**
	 * @return the timeout
	 */
	public long getTimeout() {
		return timeout;
	}

	public boolean isSendSuccess(){
		return this.sendTime > 0;
	}

	private static class TimeoutRequestScanner implements Runnable {

	        public void run() {
	            while (true) {
	                try {
	                    for (DoveFuture ft : future.values()) {
	                        if (ft == null || ft.isDone() || ft.getStartTime() <= 0) {
	                            continue;
	                        }
	                        if (System.currentTimeMillis() - ft.getStartTime() > ft.getTimeout()*30) {
	                            // create exception response.
	                            RPCReponse timeoutResponse = new RPCReponse(ft.getSeqId());
	                            // set timeout status.
//	                            timeoutResponse.setStatus(future.isSendSuccess() ? Response.SERVER_TIMEOUT : Response.CLIENT_TIMEOUT);
	                            timeoutResponse.setStatus(false);
//	                            timeoutResponse.setErrorMessage(future.getTimeoutMessage(true));
	                            // handle response.
	                            timeoutResponse.setMessage("rpc request or server response time out exception.");
	                            DoveFuture.readResponse(timeoutResponse);
	                        }
	                    }
	                    Thread.sleep(60000);
	                } catch (Throwable e) {
	                    logger.error("Exception when scan the timeout invocation of remoting.", e);
	                }
	            }
	        }
	    }

	    static {
	        Thread th = new Thread(new TimeoutRequestScanner(), "DoveResponseTimeoutScaner");
	        th.setDaemon(true);
	        th.start();
	    }
	
}
